package cn.nawang.ebeim.test.transfer;

import cn.nawang.ebeim.test.constants.Config;
import cn.nawang.ebeim.test.service.TestData;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpContentDecompressor;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.SystemPropertyUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

public class TransferClient {

    private static final Logger LOG = LoggerFactory.getLogger(TransferClient.class);

    /**
     * 为了复用EventLoopGroup 、Selector等nio资源，client维持单例,
     * 所有的传输任务,只是新建链路channel,并由netty注册到Selector上
     */
    private static TransferClient transferClient;

    private static int CPU_NUM;

    static {
        CPU_NUM = Runtime.getRuntime().availableProcessors();
    }

    /**
     * 多长时间未请求后，发送心跳
     */
    private static final int WRITE_WAIT_SECONDS = 5;

    private EventLoopGroup eventLoopGroup;

    private TransferClient() {
        TestData.CUP_NUM = CPU_NUM;
        TestData.setConcurrentCustomerNum(Config.CONCURRENT_NUM);
        eventLoopGroup = new NioEventLoopGroup(Config.CONCURRENT_NUM);
    }

    public static TransferClient getTransferClient() {
        if (transferClient == null) {
            synchronized (TransferClient.class) {
                if (transferClient == null) {
                    transferClient = new TransferClient();
                }
            }
        }
        return transferClient;
    }

    public void close() {
        if (eventLoopGroup != null && !eventLoopGroup.isShuttingDown()) {
            try {
                eventLoopGroup.shutdownGracefully().sync();
                LOG.info("EventLoopGroup shutdown complete!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void ensureUsable() {
        if (eventLoopGroup == null || eventLoopGroup.isShutdown()) {
            eventLoopGroup = new NioEventLoopGroup();
            LOG.info("EventLoopGroup create!");
        }
    }

    public void download(FileInfo info, String host, int port,TransferCallback callback) throws TransferException {
        ensureUsable();
        final TransferDownloadHandler handler = new TransferDownloadHandler(info,callback);
        try {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("codec", new HttpClientCodec());
                            pipeline.addLast("ping", new IdleStateHandler(0, WRITE_WAIT_SECONDS, 0, TimeUnit.SECONDS));
                            pipeline.addLast("inflater", new HttpContentDecompressor());
                            pipeline.addLast("chunkWriter", new ChunkedWriteHandler());
                            pipeline.addLast("handler", handler);
                        }
                    });
            b.connect(host, port).sync();
        }catch (Exception e) {
            throw new TransferException("dsId: " + info.getDataId() + " signature: " + info.getSignature() + ", version: " + info.getVersion() + "==>>[" + host + "]下载失败: "+ handler.getErrorMsg());
        }
    }

    public void upload(FileInfo info, String host, int port,TransferCallback callback) throws TransferException {
        ensureUsable();
        final TransferUploadHandler handler = new TransferUploadHandler(info,callback);
        try {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup).channel(NioSocketChannel.class);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("codec", new HttpClientCodec());
                    pipeline.addLast("http-encoder", new ChunkedWriteHandler());
                    pipeline.addLast("ping",new IdleStateHandler(0, WRITE_WAIT_SECONDS, 0, TimeUnit.SECONDS));
                    pipeline.addLast("dispatcher", handler);
                }
            });
            b.connect(host, port).sync();
        } catch (Exception e) {
            throw new TransferException("dsId: " + info.getDataId() + " signature: " + info.getSignature() + ", version: " + info.getVersion() + "==>>[" + host + "]上传失败: "+ handler.getErrorMsg());
        }
    }

}
